package com.cloudcanal.processor;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import javax.sql.DataSource;

import com.clougence.cloudcanal.sdk.api.modelv2.CustomData;
import com.clougence.cloudcanal.sdk.api.modelv2.CustomFieldV2;
import com.clougence.cloudcanal.sdk.api.modelv2.CustomRecordV2;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public abstract class DataChangeModify implements RecordModifyConvert{
	
	public List<CustomData> insert(CustomData data, DataSource srcDataSource){
		List<CustomRecordV2> records = new ArrayList<CustomRecordV2>();
		for (CustomRecordV2 recordV2 : data.getRecords()) {
			Map<String, CustomFieldV2> fieldMapAfter = recordV2.getAfterColumnMap();
			DataChangePo changePo = query(srcDataSource, Long.valueOf(fieldMapAfter.get("id").getValue().toString()));
			if (changePo != null) {
				recordV2.addField(CustomFieldV2.buildField("payload", changePo.getPayload(), Types.VARCHAR, false, false, true));
				recordV2.addField(CustomFieldV2.buildField("data_source", changePo.getData_source(), Types.VARCHAR, false, false, true));
				records.add(recordV2);
			}
		}
		return Arrays.asList(new CustomData(data.getSchemaInfo(), data.getEventType(), records));
	}
	
	public int updateDataChange(DataChangePo changePo, DataSource dstDataSource) {
		Connection connection = null;
		try {
			connection = dstDataSource.getConnection();
			PreparedStatement ps = connection
					.prepareStatement("update data.data_log set payload = ? , is_handle = 0 where data_source = ? and src_id = ? ");
			ps.setString(1, changePo.getPayload());
			ps.setString(2, changePo.getData_source());
			ps.setLong(3, changePo.getSrcId());
			return ps.executeUpdate();
		} catch (Exception e) {
			throw new RuntimeException(e);
		} finally {
			if (connection != null) {
				try {
					connection.close();
				} catch (SQLException e) {
				}
			}
		}
	}
	
	public void update(CustomData data, DataSource srcDataSource, DataSource dstDataSource) {
		for (CustomRecordV2 recordV2 : data.getRecords()) {
			Map<String, CustomFieldV2> fieldMapAfter = recordV2.getAfterColumnMap();
			Long srcId = Long.valueOf(fieldMapAfter.get("id").getValue().toString());
			DataChangePo changePo = query(srcDataSource, srcId);
			if (changePo != null) {
				updateDataChange(changePo, dstDataSource);
			}
		}
	}
	
	@Override
	public List<CustomData> modify(CustomData data, DataSource srcDataSource, DataSource dstDataSource) {
		switch (data.getEventType()) {
		case INSERT:
			return insert(data, srcDataSource);
		case UPDATE:
			update(data, srcDataSource, dstDataSource);
			break;
		default:
			break;
		}
		return new ArrayList<CustomData>();
	}
	
	public abstract String getPrepareStatementSql();
	
	public DataChangePo query(DataSource dataSource, Long srcId) {
		Connection connection = null;
		try {
			connection = dataSource.getConnection();
			PreparedStatement ps = connection.prepareStatement(getPrepareStatementSql());
			ps.setLong(1, srcId);
			ResultSet resultSet = ps.executeQuery();
			if (resultSet.next()) {
				DataChangePo changePo = DataChangePo.builder().data_source(resultSet.getString("data_source")).payload(resultSet.getString("payload")).srcId(srcId).build();
				return changePo;
			}
			return null;
		} catch (Exception e) {
			log.error("query error", e);
			throw new RuntimeException(e);
		} finally {
			if (connection != null) {
				try {
					connection.close();
				} catch (SQLException e) {
				}
			}
		}
	}
}
